package com.huan.kafka.producer;

import jakarta.annotation.Resource;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 消息生产者
 */
@Component
public class KafkaProducer implements CommandLineRunner {

    public static final String TOPIC = "topic-001";
    private static final Logger log = LoggerFactory.getLogger(KafkaProducer.class);

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void run(String... args) {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() ->
                {
                    kafkaTemplate.send(TOPIC, "当前时间:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
                            .whenComplete((sendResult, throwable) -> {
                                RecordMetadata recordMetadata = sendResult.getRecordMetadata();
                                if (null != recordMetadata) {
                                  //  log.info("消息发送成功 offset: {}", recordMetadata.offset());
                                }
                                if (null != throwable) {
                                //    log.error("消息发送失败，错误信息。", throwable);
                                }
                            });
                },
                0, 1, TimeUnit.MILLISECONDS);
    }
}
